package com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqSerializeException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.MessageType;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.serializer.Serializer;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.rabbitmq.client.AMQP;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;

@Slf4j
public class DefaultSerializerProvider implements SerializerProvider {

    @Data
    @AllArgsConstructor
    static class Key {

        private String contentType;
        private String contentEncoding;
        private Class targetClass;
    }

    private static LoadingCache<Key, Serializer> serializerCache = CacheBuilder.newBuilder()
            .build(new CacheLoader<Key, Serializer>() {
                @Override
                public Serializer load(Key key) {
                    Serializer serializer;
                    if (key.getTargetClass() == String.class) {
                        serializer = MessageType.PLAIN.newSerializer();
                    } else {
                        serializer = MessageType.fromContentType(key.getContentType()).newSerializer();
                    }
                    if (StringUtils.isNotBlank(key.getContentEncoding())) {
                        serializer.setCharset(Charset.forName(key.getContentEncoding()));
                    } else {
                        serializer.setCharset(StandardCharsets.UTF_8);
                    }
                    return serializer;
                }
            });

    @Override
    public Serializer provide(AMQP.BasicProperties properties, Class targetClass) {
        Key key = new Key(properties.getContentType(), properties.getContentEncoding(), targetClass);
        try {
            return serializerCache.get(key);
        } catch (ExecutionException e) {
            StringBuilder stringBuilder = new StringBuilder();
            properties.appendPropertyDebugStringTo(stringBuilder);
            log.error("无法找到该消息类型的反序列化器: " + stringBuilder);
            throw new MqSerializeException("无法找到该消息类型的反序列化器: " + stringBuilder, null, null, e);
        }
    }
}
